package com.abyss.szse;


import com.abyss.avro.SzseAvro;
import com.abyss.kafka.KafkaPro;

import java.io.*;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.Socket;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * @Date 2020/9/12
 */
public class SocketClient {

    //随机浮动成交价格系数
    private static Double[] price = new Double[]{0.1, 0.11, 0.12, 0.13, 0.14, 0.15, 0.16, 0.17, 0.18, 0.19, 0.2, -0.1, -0.11, -0.12, -0.13, -0.14, -0.15, -0.16, -0.17, -0.18, -0.19, -0.2};

    //随机浮动成交量
    private static int[] volumn = new int[]{50, 80, 110, 140, 170, 200, 230, 260, 290, 320, 350, 380, 400, 500, 600, 700, 800, 900, 1000, 1100, 1200, 1300};
    static Map<String, Map<String,Long>> mapCache ;
    public static void main(String[] args) throws Exception {
        /**
         * 开发步骤:
         * 1.创建main方法
         * 2.建立socket连接，获取流数据
         * 3.读文件缓存成交量和成交金额
         * 4.解析行数据，数据转换
         * 5.发送kafka
         */
        //2.建立socket连接，获取流数据
        Socket socket = new Socket("localhost", 4444);
        InputStream inputStream = socket.getInputStream();
        DataInputStream ds = new DataInputStream(inputStream);

        //3.读文件缓存成交量和成交金额
        mapCache = cacheFile();
        KafkaPro kafkaPro = new KafkaPro();
        while (true){
            //获取每一条数据
            String str = ds.readUTF();
            //4.解析行数据，数据转换
            SzseAvro szseAvro = trannsfer(str);
            System.out.println(szseAvro);
            //5.发送kafka
            kafkaPro.send("szse",szseAvro);
        }
    }

    /**
     * 解析行数据，数据转换
     * @param str
     */
    private static SzseAvro trannsfer(String str) {
        /**
         * 1.拷贝成交量和成交价数组
         * 2.获取随机浮动成交量和价格
         * 3.字符串切割、计算最新价
         * 4.获取缓存的成交量/金额
         * 5.计算总成交量/金额
         * 6.缓存总成交量/金额
         * 7.获取最高价和最低价(和最新价比较)
         * 8.封装结果数据
         */
        //2.获取随机浮动成交量和价格
        Random random = new Random();
        int pIndex = random.nextInt(price.length);
        Double priceRandom = price[pIndex];//随机价格系数
        //获取浮动成交量
        int vIndex = random.nextInt(volumn.length);
        int vol = volumn[vIndex];
        //3.字符串切割、计算最新价
        String[] arr = str.split("\\|");
        String code = arr[1].trim();
        BigDecimal tradePrice = new BigDecimal(arr[9].trim());
        tradePrice = tradePrice.multiply(new BigDecimal(1+priceRandom)).setScale(2,RoundingMode.HALF_UP);
        //4.获取缓存的成交量/金额
        Map<String, Long> volAmtMap = mapCache.get(code);
        Long tradeVolDay = 0L; //日总成交量
        Long tradeAmtDay = 0L; //日总成交金额
        if(volAmtMap != null){
            //5.计算总成交量/金额
            Long tradeVol = volAmtMap.get("tradeVol");
            Long tradeAmt = volAmtMap.get("tradeAmt");
            tradeVolDay = tradeVol + vol;
            //当前这一s钟的成交金额
            BigDecimal tmpAmt = tradePrice.multiply(new BigDecimal(vol)).setScale(2, RoundingMode.HALF_UP);
            tradeAmtDay = tradeAmt+tmpAmt.longValue();
            //6.缓存总成交量/金额
            volAmtMap.put("tradeVol",tradeVolDay);
            volAmtMap.put("tradeAmt",tradeAmtDay);
            mapCache.put(code,volAmtMap);
        }

        // 7.获取最高价和最低价(和最新价比较)
        BigDecimal highPrice = new BigDecimal(arr[7].trim());
        BigDecimal lowPrice = new BigDecimal(arr[8].trim());
        if(highPrice.compareTo(tradePrice) == -1){
            highPrice = tradePrice;
        }
        if(lowPrice.compareTo(tradePrice) == 1){
            lowPrice = tradePrice;
        }

        //8.封装结果数据
        SzseAvro szseAvro = new SzseAvro();
        szseAvro.setMdStreamID(arr[0].trim());
        szseAvro.setSecurityID(code);
        szseAvro.setSymbol(arr[2].trim());
        szseAvro.setTradeVolume(tradeVolDay);
        szseAvro.setTotalValueTraded(tradeAmtDay);
        szseAvro.setPreClosePx(Double.valueOf(arr[5].trim()));
        szseAvro.setOpenPrice(Double.valueOf(arr[6].trim()));
        szseAvro.setHighPrice(highPrice.doubleValue());
        szseAvro.setLowPrice(lowPrice.doubleValue());
        szseAvro.setTradePrice(tradePrice.doubleValue());
        szseAvro.setClosePx(tradePrice.doubleValue());
        szseAvro.setTradingPhaseCode("T01");
        szseAvro.setTimestamp(new Date().getTime());
        return szseAvro;
    }

    private static Map<String, Map<String, Long>> cacheFile() throws Exception {
        Map<String, Map<String, Long>> map = new HashMap<>();
        /**
         * 1.读取本地文件（深市）
         * 2.解析行数据获取成交量和成交金额
         * 3.封装数据到map
         * 4.关流
         */
        //1.读取本地文件（深市）
        BufferedReader brIndex = new BufferedReader(new InputStreamReader(new FileInputStream(new File("/Users/abyss/Dev/toys/flink/flink-stock/producer/src/main/resources/szse-index.txt"))));
        String strIndex;
        while ((strIndex = brIndex.readLine())!= null){
            String[] arr = strIndex.split("\\|");
            String code = arr[1].trim();
            Long tradeVol = Long.valueOf(arr[3].trim());//成交量
            long tradeAmt = Double.valueOf(arr[4]).longValue();//成交金额
            Map<String,Long> mapVolAmt = new HashMap<>();
            mapVolAmt.put("tradeVol",tradeVol);
            mapVolAmt.put("tradeAmt",tradeAmt);
            map.put(code,mapVolAmt);//封装map缓存数据：缓存成交量和成交金额
        }
        brIndex.close();

        BufferedReader brStock = new BufferedReader(new InputStreamReader(new FileInputStream(new File("/Users/abyss/Dev/toys/flink/flink-stock/producer/src/main/resources/szse-stock.txt"))));
        String strStock;
        while ((strStock = brStock.readLine())!= null){
            String[] arr = strStock.split("\\|");
            String code = arr[1].trim();
            Long tradeVol = Long.valueOf(arr[3].trim());//成交量
            long tradeAmt = Double.valueOf(arr[4]).longValue();//成交金额
            Map<String,Long> mapVolAmt = new HashMap<>();
            mapVolAmt.put("tradeVol",tradeVol);
            mapVolAmt.put("tradeAmt",tradeAmt);
            map.put(code,mapVolAmt);//封装map缓存数据：缓存成交量和成交金额
        }
        brStock.close();

        return map;
    }
}